#include "gtest/gtest.h"
#include <mutex>
#include <condition_variable>
#include <thread>
#include <atomic>
#include <chrono>

#include "stream/Topic.h"
#include "stream/Appender.h"
#include "stream/Reader.h"

using namespace stream;


//1. Single Appender Only 
// see: TestAppender.cpp: TEST(AppenderTest, test)
//2. Single Reader Only 
// see: TestReaderWrapper.cpp: TEST_F(ReaderWrapperTest, testSequenceReader)

//3. Single Appender (thread) & Single Reader (thread)

class State {
public:
	bool isRunning() const{
		return _state.load();
	}
	bool isStoped() const {
		return !isRunning();
	}
	void stop() {
		_state.store(false, std::memory_order_seq_cst);
	}
private:
	std::atomic<bool> _state = true;
};

class Signal {
public:
	enum class WaitState {
		Timeout,
		NoTimeout,
	};

	Signal(){
	}
	~Signal(){
	}
	WaitState wait(uint64_t ms){
		std::unique_lock lock(_mutex);
		if(std::cv_status::timeout == _signal.wait_for(lock, std::chrono::milliseconds(ms))){
			return WaitState::Timeout;
		}
		else {
			return WaitState::NoTimeout;
		}
	}
	void signal(){
		return _signal.notify_all();
	}
	std::mutex & mutexRef(){
		return _mutex;
	}
	
private:
	std::condition_variable _signal;
	std::mutex _mutex;
};


class App {
public:
	App(const std::string & dbname, const std::string & dir)
	: _topic(::stream::makeTopic(dbname, dir))
    , _reader(::stream::makeReader(_topic, 0))
	, _appender(::stream::makeAppender(_topic))
	{
		reader()->open();
		appender()->open();
	}
	~App(){
		appender()->close();
		reader()->close();	
	}

	void append(){
		printf("start appender...\n");

		for(size_t count = 0; count < 100; ++count){
			if(count % 10 == 0){
				std::this_thread::sleep_for (std::chrono::milliseconds(10));
				printf("appended thread sleep 100ms: count:%zu\n", count);
			}

			std::unique_lock lock(signal.mutexRef());
			appender()->append(count, "GTP" + std::to_string(count+1));
			signal.signal();

			printf("append GTP--%zu\n", count+1);
		}

		state.stop();

		printf("append finished\n");
	}

	void read(){
		while(true){
			while(reader()->hasNext()){
				printf("read :%s\n", reader()->current()->c_str());
				reader()->next();
			}

			if(state.isStoped()) break;

			if(Signal::WaitState::Timeout == signal.wait(20))
				printf("wait_for dataAppended event Timeout\n");
			else
				printf("wait_for dataAppended event Fired\n");
		}

		printf("reader finished\n");
	}

	std::shared_ptr<Reader> reader() {
		return _reader;
	}
	std::shared_ptr<Appender> appender(){
		return _appender;
	}

private:
	State state;
	Signal signal;

	std::shared_ptr<Topic> _topic;
	std::shared_ptr<Reader> _reader;
	std::shared_ptr<Appender> _appender;
};


TEST(FunctionTest, testAppenderReader){
	App app("hello-function.test", ".");

    //2. start append thread : append 100 count with signal read thread ,finally stop the read loop
    std::thread thread(&App::append, &app);

    //3. read loop: exit-condition:  isRunning ==false && reader can not read data
	app.read();

    //4. resource clear
    thread.join(); 
}

